/**
 *     This file is part of cesm-ws.
 *
 *     cesm-ws, Web Service for Community Earth System Model
 *
 *     Copyright (c) 2010, Purdue University and/or affilicates or thrid-party contributors as
 *     indicated by the @author tags or express copyright attribution statements
 *     applied by the authors. All third-party contributions are distributed under
 *     license by Purdue University.
 *
 *     cesm-ws is free software: you can redistribute it and/or modify
 *     it under the terms of the GNU General Public License as published by
 *     the Free Software Foundation, either version 3 of the License, or
 *     (at your option) any later version.
 *
 *     This program is distributed in the hope that it will be useful,
 *     but WITHOUT ANY WARRANTY; without even the implied warranty of
 *     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *     GNU General Public License for more details.
 *
 *     You should have received a copy of the GNU Lesser General Public License
 *     along with this distribution.  If not, see <http://www.gnu.org/licenses/>.
 */

package edu.purdue.rcac.cesm.submitter;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.regex.Matcher;

import javax.activation.DataHandler;

import org.apache.log4j.Logger;
import org.globus.gsi.GlobusCredentialException;
import org.globus.rsl.NameOpValue;
import org.ietf.jgss.GSSException;

import edu.purdue.rcac.cesm.Constants;
import edu.purdue.rcac.cesm.Directories;
import edu.purdue.rcac.cesm.util.CESMLogElement;
import edu.purdue.rcac.cesm.util.CESMLogParser;
import edu.purdue.rcac.cesm.util.StringUtil;
import edu.purdue.rcac.grid.submit.SubmissionException;
import edu.purdue.rcac.grid.submit.Submitter;
import edu.purdue.rcac.grid.submit.SubmitterResponse;
import edu.purdue.rcac.grid.submit.SubmitterResponse.Outcome;
import edu.purdue.rcac.grid.submit.globus.GlobusSubmitter;
import edu.purdue.rcac.grid.submit.globus.RSLAttribute;
import edu.purdue.rcac.grid.util.ByteArrayDataSource;

public class CESMGlobusSubmitter extends GlobusSubmitter implements Submitter {

	public CESMGlobusSubmitter() throws GlobusCredentialException, GSSException {
		super();
	}

	public CESMGlobusSubmitter(HashMap<String, String> argsMap)
			throws GlobusCredentialException, GSSException {
		this();
		this.setArgsMap(argsMap);
	}

	protected HashMap<String, String> argsMap;

	protected static final Logger logger = Logger
			.getLogger(CESMGlobusSubmitter.class.getName());

	private static final String TG_STEELE_NAME = "tg-steele.purdue.teragrid.org";
	private static final String CCSMUSER_ALLOCATION = "TG-ATM090060";
	private static final int DATA_SERVER_PORT = 2811;
	private static final String TG_DATA_ADDR = "tg-data.purdue.teragrid.org";
	private static final ArrayList<String> FTP_OP_LIST = new ArrayList<String>(
			Arrays.asList(new String[] { "list", "mkdir", "remove", "move",
					"rename" }));

	private static final int BLOCK_SIZE = 2048;

	private String jobDescription = "";
	private String message = "";
	private String error = "";

	private String username;
	private String casename;
	private String action;

	private CESMLogParser logs;

	public void setArgsMap(Map<String, String> argsMap) {
		this.argsMap = (HashMap<String, String>) argsMap;

		for (String key : argsMap.keySet()) {
			logger.info(key + "=>" + argsMap.get(key));
		}

		String params = "";
		String outPrefix = "";
		String targetScript = "";
		action = StringUtil.underscoreLowerCase(argsMap
				.get(Constants.ParamName.ACTION.name()));
		username = argsMap.get(Constants.ParamName.USERNAME.name());
		casename = argsMap.get(Constants.ParamName.CASENAME.name());

		Iterator it = argsMap.keySet().iterator();
		while (it.hasNext()) {
			String key = it.next().toString();
			if (key.startsWith(Constants.PARAM + Constants.DELIMITOR)) {
				// action param
				String param = Constants.unwrapParam(key);
				params += (param + "=" + argsMap.get(key) + ",");
				it.remove();
			}

		}
		outPrefix = Directories.CESM_GLOBUS_LOG_DIR + File.separator;
		if ("update_all_runtimes".equals(action)) {
			// update
			outPrefix += "update" + File.separator + "update-all";
		} else if ("track_resubmit_pbsid".equals(action)) {
			// track resubmit pbsid
			outPrefix += "update" + File.separator + "track_resubmit_pbsid";
		} else if ("create_case".equals(action)) {
			// create
			outPrefix += "create" + File.separator + username + File.separator
					+ casename + ".create";
		} else if ("delete_case".equals(action)) {
			// delete
			outPrefix += "delete" + File.separator + username + File.separator
					+ casename + ".delete";
		} else if ("create_user_home".equals(action)) {
			// create user home
			outPrefix += "create_user_home" + File.separator + username;
		} else if (FTP_OP_LIST.contains(action)) {
			// ftp operation
			outPrefix += "ftp" + File.separator + username + File.separator
					+ action;
		} else if ("".equals(casename) || casename == null
				|| action.equals("get_case_name_list")) {
			outPrefix += username + File.separator + action;
		} else {
			outPrefix += username + File.separator + casename + File.separator
					+ action;
		}
		Timestamp ts = new Timestamp(System.currentTimeMillis());
		outPrefix += "." + ts.toString();
		logger.info("stdout: " + outPrefix + ".gout");
		logger.debug("stdout: " + outPrefix + ".gout");

		targetScript = Directories.CESM_WSSCRIPTS_DIR + File.separator
				+ StringUtil.underscoreLowerCase(action) + ".sh";

		if (params.length() != 0) {
			params = params.substring(0, params.length() - 1);
			String[] args = new String[] { "-e", params, "-v" };
			addNameOpValue(RSLAttribute.ARGUMENTS, NameOpValue.EQ, args);
		}

		addNameOpValue(RSLAttribute.STDOUT, NameOpValue.EQ, outPrefix + ".gout");
		addNameOpValue(RSLAttribute.STDERR, NameOpValue.EQ, outPrefix + ".gerr");
		addNameOpValue(RSLAttribute.EXECUTABLE, NameOpValue.EQ, targetScript);
		addNameOpValue(RSLAttribute.PROJECT, NameOpValue.EQ,
				CCSMUSER_ALLOCATION);

		this.jobDescription = translate();
	}

	public String getError() {
		return error;
	}

	public String getMessage() {
		return message;
	}

	public String getJobDescription() {
		return jobDescription;
	}

	private CESMLogElement getLatestLog() {
		try {
			logs = new CESMLogParser(this.getOutContent());
		} catch (Exception e) {
			e.printStackTrace();
			return null;
		}
		return logs.getLatestLog();
	}

	private String getLatestGlobusErr() {
		try {
			error = this.getErrContent();
			Matcher matcher = CESMLogParser.logPattern.matcher(error);
			if (matcher.find()) {
				error = matcher.group(2);
			}
		} catch (Exception e) {
			logger.error(e);
			error = null;
			return null;
		}
		return error;
	}

	public void submit() throws SubmissionException {
		try {
			String rsl = this.jobDescription;

			logger.info("RSL: \n\t" + rsl.replaceAll("[)]", ")\n\t"));
			logger.debug("RSL: \n\t" + rsl.replaceAll("[)]", ")\n\t"));
			this.submit(TG_STEELE_NAME, rsl);

		} catch (Exception e) {
			throw new SubmissionException(e);
		}
	}

	public SubmitterResponse getResult() {
		SubmitterResponse response = new SubmitterResponse();
		CESMLogElement latestLog = getLatestLog();
		logger.debug("LATEST LOG: " + latestLog);
		getLatestGlobusErr();
		if (latestLog != null) {
			if (latestLog.isErrorOccured()) {
				response.setOutcome(Outcome.FAIL);
				response.setError("Error caught via stdout:\n"
						+ latestLog.getError());
				logger.info(latestLog.getError());
			} else {
				response.setOutcome(Outcome.SUCCESS);
				response.setResult(latestLog.getResult());
			}
			response.setLog(latestLog.getContent());
		} else {
			response.setError("I/O Exception for log");
		}
		if (error != null) {
			response.setError(response.getError()
					+ "Error caught via stderr:\n" + error);
		} else {
			response.setError("I/O Exception for err");
		}
		return response;
	}

	public SubmitterResponse write(DataHandler data, String toPath,
			boolean pathIsRelative) throws IOException {
		String path = toPath;
		for (String key : argsMap.keySet()) {
			logger.info(key + " -> " + argsMap.get(key));
		}
		if (pathIsRelative) {
			path = Directories.CESM_UPLOAD_DIR + File.separator
					+ argsMap.get(Constants.ParamName.USERNAME.name())
					+ File.separator + toPath;
		}
		OutputStream output = this.getOutputStream(path, TG_DATA_ADDR,
				DATA_SERVER_PORT, false);
		logger.info("Globus: write to " + path);
		byte[] buffer = new byte[4096];
		InputStream is = data.getInputStream();
		int num = is.read(buffer);
		while (num != -1) {
			output.write(buffer, 0, num);
			System.out
					.println(String.format("[%d] %s", num, buffer.toString()));
			num = is.read(buffer);
		}
		output.close();
		return new SubmitterResponse(Outcome.SUCCESS);
	}

	public ByteArrayDataSource readAsByteArray(String fromPath,
			boolean pathIsRelative) throws IOException {
		// TODO: deal with large files
		InputStream is = this.getInputStream(fromPath, TG_DATA_ADDR,
				DATA_SERVER_PORT);
		logger.info("Globus: read from " + fromPath);
		byte[] bytes = new byte[BLOCK_SIZE];
		ByteArrayOutputStream baos = new ByteArrayOutputStream();
		int offset = 0, numRead = 0;
		while ((numRead = is.read(bytes, 0, BLOCK_SIZE)) != -1) {
			// System.out.println("num: "+numRead+" | offset: " +offset);
			baos.write(bytes, 0, numRead);
			offset += BLOCK_SIZE;
		}
		logger.info("read " + baos.size() + " bytes");
		return new ByteArrayDataSource(baos.toByteArray(), baos.size(),
				"application/octet-stream");
	}

	public String readAsString() throws IOException {
		String action = argsMap.get(Constants.ParamName.ACTION.name());
		String path = "";
		if (action != null && action.equals("get_CaseStatus")) {
			String username = argsMap.get(Constants.ParamName.USERNAME.name());
			String casename = argsMap.get(Constants.ParamName.CASENAME.name());

			path = Directories.CESM_CASE_DIR + File.separator + username
					+ File.separator + casename + File.separator + "CaseStatus";
		} else {
			String fileRelativePath = argsMap.get(Constants.ParamName.FILE
					.name());
			path = Directories.CESM_GLOBUS_LOG_DIR + File.separator
					+ fileRelativePath;
		}
		return this.getFileContent(TG_DATA_ADDR, DATA_SERVER_PORT, path);
	}
}
